package com.ww.flink

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessAllWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.table.sources.wmstrategies.BoundedOutOfOrderTimestamps
import org.apache.flink.util.Collector


/**
 * 设置
 */
object Flink_try16_allowedlateness {
  case class CarInfo(carId: String, speed: Long)

  def main(args: Array[String]): Unit = {
    //定义了测输出流的标签
    val tag = new OutputTag[(Long, String)]("late")
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //设置时间类型为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val initStream: DataStream[String] = env.socketTextStream("node01", 8888)
    //把字符串时间转换成long
    val mapStream: DataStream[(Long, String)] = initStream.map(_.split(" ")).map(arr => (arr(0).toLong, arr(1)))
    //设置watermark 延迟2秒
    val watermarksStream: DataStream[(Long, String)] = mapStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[(Long, String)](Time.seconds(2)) {
      override def extractTimestamp(element: (Long, String)) = element._1
    })
    //设置窗口为 5秒
    val res: DataStream[Iterable[(Long, String)]] = watermarksStream.timeWindowAll(Time.seconds(5))
      //窗口触发之后的3s内，如果又出现了这个窗口的数据，这个窗口会重复计算  相当于说窗口保留3s
      .allowedLateness(Time.seconds(3))
      //如果数据迟到时间 超过5s 那么输出到侧输出流钟
      .sideOutputLateData(tag)
      //处理的是主流的数据，不会处理迟到非常严重的数据（已经输出到侧输出流）
      .process(new ProcessAllWindowFunction[(Long, String), Iterable[(Long, String)], TimeWindow] {
        override def process(context: Context, elements: Iterable[(Long, String)], out: Collector[Iterable[(Long, String)]]): Unit = {
          println(context.window.getStart + "---" + context.window.getEnd)
          out.collect(elements)
        }
      })
    //主输出流
    res.print("main")
    //测输出流
    res.getSideOutput(tag).print("late")
    env.execute()
  }


}
